import java.io.*;
import java.util.*;

/**
 * Encapsulates the query optimizer functionality.
 */
public class QueryOptimizer
{
    /**
     * Initializes a new instance of the QueryOptimizer class.
     *
     * @query the query as a string of terms
     * @costs the estimated costs
     */
    public QueryOptimizer(String query, EstimatedCosts costs) {
        this.query = query;
        fillSelectivities(query);

        this.r = costs.getCostOfArrayAccess();
        this.t = costs.getCostOfIfTest() ;
        this.l = costs.getCostOfLogicalAnd();
        this.m = costs.getCostOfBranchMisprediction();
        this.a = costs.getCostOfAnswerArray();
        this.f = costs.getCostOfApplyingFunction();
    }

    /**
     * Computes the optimal query plan based on the estimated costs.
     */
    public void compute() {
        // initialize a record array of size of 2^k
        int size = (int)Math.pow(2, k);
        this.array = new Record[size];

        // first step: consider all plan with no &&s
        // generating all 2^k-1 plan using only &-terms
        for (int bitIndex = 1; bitIndex < size; bitIndex++) {
            double prod_s = 1;
            double sum_f = 0;
            int n = 0;

            // check each bit, see if this particular selectivity is
            // in the current &-term
            int mask = 1;           // use to test if a specific bit is set, 0001
            for (int j = 1; j <= k; j++) {
                if ((bitIndex & mask) > 0) {
                    prod_s *= s_double[j];
                    sum_f += f;     // since f are the same, we can use f += f;
                    n++;
                }
                
                mask = mask * 2;
            }

            // compute cost based on 4.4, 4.5 and 4.7
            double costFixed = computeFixedCost(n, sum_f);
            double costNoBranch = computeNoBranchCost(n, sum_f);
            double costLogicalAnd = computeLogicalAndCost(costFixed, prod_s);

            this.array[bitIndex] = new Record();
            this.array[bitIndex].n = n;
            this.array[bitIndex].p = prod_s;
            this.array[bitIndex].fCost = costFixed;

            if (costNoBranch < costLogicalAnd) {
                this.array[bitIndex].c = costNoBranch;
                this.array[bitIndex].b = true;
            } else {
                this.array[bitIndex].c = costLogicalAnd;
            }
        }

        // step two:
        for (int s = 1; s < size; s++) {
            for (int sprime = 1; sprime < size; sprime++) {
                if ((s & sprime) == 0) {    // Check intersection
                    int leftMost = s;
                    while (array[leftMost].left != 0) {
                        leftMost = array[leftMost].left;
                    }

                    if (cMetricTest(sprime, leftMost)) {
                        // do nothing; suboptimal per Lemma 4.8 
                    } else if ((array[sprime].p <= 0.5) && (dMetricTest(sprime, s))) {
                        // do nothing; suboptimal per Lemma 4.9 
                    } else {
                        double costCombined = computeCombinedCost(sprime, s);
                        int union = sprime | s;
                        if (costCombined < this.array[union].c) {
                            this.array[union].c = costCombined;
                            this.array[union].left = sprime;
                            this.array[union].right = s;
                        }
                    }
                }
            }
        }
    }

    /**
     * Convert a single line of selectivities from the query.txt
     * to an array of doubles.
     */
    private void fillSelectivities(String query) {
        String[] s_string = this.query.split(" ");
        k = s_string.length;

        this.s_double = new double[k + 1];
        for (int i = 1; i <= k; i++) {
            s_double[i] = Double.parseDouble(s_string[i - 1]);
        }
    }

    /**
     * Prints out the results to standard output in the required format.
     */
    public void printResults() {
        System.out.println("======================================");
        System.out.println(this.query);
        System.out.println("--------------------------------------");
        
        // prepare for the output
        String innerTerm = "answer[j] = i;";
        Stack<Integer> stack = new Stack<Integer>();
        int index = this.array.length - 1;
        String answer = buildCombinedString(stack, index, "");
        String tabStop = "    ";
        
        if (stack.isEmpty()) {
            if (array[index].b == true) {
                String lastTerm = "j += " + answer + ";";
                System.out.println(innerTerm);
                System.out.println(lastTerm);
            } else {
                System.out.println("if(" + answer + ") {");
                System.out.println(tabStop + innerTerm);
                System.out.println("}");
            }
        } else {
            if (array[stack.pop()].b) {
                int lastAndIndex = answer.lastIndexOf(" && (");
                int nextRightParen = -1;
                if (lastAndIndex > 0) {
                    // In the case where the last item is && without the touching parensthesis
                    // we need to recalculate the index accordingly.
                    if (answer.substring(lastAndIndex + 3, answer.length()).contains("&&")) {
                        lastAndIndex = answer.lastIndexOf(" && ");
                        nextRightParen = answer.indexOf(")", lastAndIndex) - 1;
                    } else {
                        nextRightParen = answer.indexOf(")", lastAndIndex);
                    }
                }
                
                String lastTerm = null;
                if (nextRightParen > 0) {
                    String lastGroupTerm = answer.substring(lastAndIndex, nextRightParen + 1);
                    answer = answer.replace(lastGroupTerm, "");

                    if (lastGroupTerm.contains(" & ")) {
                        int lastLeftParen = answer.lastIndexOf("(");
                        nextRightParen = answer.indexOf(")", lastLeftParen);
                        String finalGroup = answer.substring(lastLeftParen, nextRightParen + 1);
                    
                        // Remove first and last character (i.e. trailing opening and closing parenthesis)
                        String newFinalGroup = finalGroup.substring(1, finalGroup.length() - 1);
                        answer = answer.replace(finalGroup, newFinalGroup);
                    }
                    
                    // Remove the leading group-by (&&) to get the final term
                    lastTerm = lastGroupTerm.substring(3, lastGroupTerm.length()).trim();
                }
                
                System.out.println("if" + answer +" {");
                System.out.println(tabStop + innerTerm);
                if (lastTerm != null) {
                    System.out.println(tabStop + "j += " + lastTerm + ";");
                }
                System.out.println("}");
            } else {
                System.out.println("if" + answer + " {");
                System.out.println(tabStop + innerTerm);
                System.out.println("}");
            }
        }

        System.out.println("--------------------------------------");
        System.out.println("cost = " + array[array.length - 1].c);
    }

    /**
     * Computes the fixed-cost.
     */
    private double computeFixedCost(int k, double sum_f){
        return (k * r) + ((k - 1) * l) + sum_f + t;
    }

    //example 4.5
    //(k * r) + ((k - 1) * l) + sum_f + t + (m * q) + (this.p * a);
    private double computeLogicalAndCost(double fCost, double prod_p) {
        if (prod_p <= 0.5) {
            return fCost + (m * prod_p) + (prod_p * a);
        }

        return fCost + (m * (1 - prod_p)) + (prod_p * a);
    }

    // example 4.4
    private double computeNoBranchCost(int k, double sum_f) {
        return (k * r) + ((k - 1) * l) + sum_f + a;
    }

    //Eq. (1)
    private double computeCombinedCost(int sprime, int s) {
        double p = array[sprime].p;
        return (array[sprime].fCost + m * ((p <= 1 - p) ? p : 1 - p) + p * array[s].c);
    }

    /**
     * Calculates the c-metric and tests whether the condition has been satisfied.
     */
    private boolean cMetricTest(int sprime, int leftMost) {
        return 
            (array[leftMost].p <= array[sprime].p) && 
            (((array[leftMost].p - 1) / array[leftMost].fCost) < ((array[sprime].p - 1) / array[sprime].fCost));
    }

    /**
     * Calculates the d-metric and tests whether the condition has been satisfied.
     */
    private boolean dMetricTest(int sprime, int s) {
        Queue<Integer> qe = new LinkedList<Integer>();
        getAllNTerms(qe, s);

        boolean result = false;
        while (!qe.isEmpty()) {
            int right = qe.remove();
            if ((array[right].p < array[sprime].p) && (array[right].fCost < array[sprime].fCost)) {
                result = true;
                break;
            }
        }
			
        return result;
    }

    /**
     * Gets all the n-terms in a combined term.
     */
    private void getAllNTerms(Queue<Integer> qe, int s){
        if (array[s].left == 0) {
            qe.add(s);
            return;
        }

        getAllNTerms(qe, array[s].left);
        getAllNTerms(qe, array[s].right);
    }

    /**
     * Gets all n-terms in a combined term.
     */
    private String buildCombinedString(Stack<Integer> stack, int index, String answer) {
        if (array[index].left == 0) {
            return answer + buildNtermString(index);
        }

        stack.push(index);

        StringBuilder temp = new StringBuilder(answer);
        temp.append("(");
        temp.append(buildCombinedString(stack, array[index].left, answer));
        temp.append(" && ");
        temp.append(buildCombinedString(stack, array[index].right, answer));
        temp.append(")");

        return temp.toString();
    }

    /**
     * precondition index > 0
     * construct &-term output string
     * example 1:  t1[o1[i]]
     * example 2:  (t1[o1[i]] & t5[o5[i]])
     */
    private String buildNtermString(int index){        
        int mask = 1;
        String terms = "";
        int numOfTerms = 0;
        for (int j = 1; j <= k; j++) {
            if ((index & mask) > 0) {
                terms +=  "t" + j + "[o" + j + "[i]] & ";
                numOfTerms++;
            }

            mask = mask * 2;
        }

        // We remove the last three characters for the trailing & 
        String result = terms.substring(0, terms.length() - 3);
        return (numOfTerms > 1) ? "(" + result + ")" : result;
    }

    private String query;
    private double[] s_double;
    private int k;
    private int r;
    private int t;
    private int l;
    private int m;
    private int a;
    private int f;
    private Record[] array;
}